package org.xxd.kafka.clients.common.record;

import java.nio.ByteBuffer;
import java.util.Iterator;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-09-27 17:28
 * @description: 内存实际存储的数据结构
 */
public class MemoryRecords implements Records {
    private final static int WRITE_LIMIT_FOR_READABLE_ONLY = -1;

    // the compressor used for appends-only
    //  private final Compressor compressor;

    // the write limit for writable buffer, which may be smaller than the buffer capacity
    private final int writeLimit;

    // the capacity of the initial buffer, which is only used for de-allocation of writable records
    private final int initialCapacity;

    // the underlying buffer used for read; while the records are still writable it is null
    private ByteBuffer buffer;

    // indicate if the memory records is writable or not (i.e. used for appends or read-only)
    // 代表这个records 是否可用写入
    private boolean writable;

    // Construct a writable memory records
    private MemoryRecords(ByteBuffer buffer, CompressionType type, boolean writable, int writeLimit) {
        this.writable = writable;
        this.writeLimit = writeLimit;
        this.initialCapacity = buffer.capacity();
        if (this.writable) {
            this.buffer = null;
            // this.compressor = new Compressor(buffer, type);
        } else {
            this.buffer = buffer;
            // this.compressor = null;
        }
    }


    public static MemoryRecords emptyRecords(ByteBuffer buffer, CompressionType type, int writeLimit) {
        return new MemoryRecords(buffer, type, true, writeLimit);
    }

    public static MemoryRecords emptyRecords(ByteBuffer buffer, CompressionType type) {
        // use the buffer capacity as the default write limit
        return emptyRecords(buffer, type, buffer.capacity());
    }

    public static MemoryRecords readableRecords(ByteBuffer buffer) {
        return new MemoryRecords(buffer, CompressionType.NONE, false, WRITE_LIMIT_FOR_READABLE_ONLY);
    }

    @Override
    public int sizeInBytes() {
        return buffer.limit();
    }

    /**
     * Get the byte buffer that backs this records instance for reading
     */
    public ByteBuffer buffer() {
        if (writable)
            throw new IllegalStateException("The memory records must not be writable any more before getting its underlying buffer");

        return buffer.duplicate();
    }

    @Override
    public Iterator<LogEntry> iterator() {
        return null;
    }

    public int initialCapacity() {
        return this.initialCapacity;
    }

    public void close() {
        if (writable) {
            // close the compressor to fill-in wrapper message metadata if necessary
            // compressor.close();

            // flip the underlying buffer to be ready for reads
            // buffer = compressor.buffer();
            // buffer.flip();

            // reset the writable flag
            writable = false;
        }
    }

    public boolean isFull() {
        return !this.writable;
    }
}
